package utils;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;

/**
 * @auther: wuyahan
 * @Date: 2021/4/13 11:44
 * @Description:
 */
public class JsonUtils {

    private static final Logger logger = LoggerFactory.getLogger(JsonUtils.class);


    /**
     * kafka数据校验
     *
     * @param data 数据
     * @return
     */
    public static boolean kafkaStreamDataCheck(String data, List<String> dataKeys, List<String> dataTypes) {
        try {
            System.out.println("数据校验格式为json");
            //数据json方式处理
            JSONObject object = JSON.parseObject(data);
            //TODO 针对每个数据进行数据类型校验(待完善)
            for (int i = 0; i < dataKeys.size(); i++) {
                Object dataObject = object.get(dataKeys.get(i));
                FlinkDataType.transformValue(FlinkDataType.getTypeInformation(dataTypes.get(i)).getTypeClass(),(String) dataObject);

            }
        } catch (Exception e) {
            System.out.println("json数据==" + data);
            System.out.println("json数据解析错误");
            return false;

        }

        return true;

    }

    /**
     * 校验ogg格式的kafka数据
     *
     * @param data
     * @param dataKeys
     * @param dataTypes
     * @return
     */
    public static boolean kafkaOggStreamDataCheck(String data, List<String> dataKeys, List<String> dataTypes) {
        try {
            logger.info("数据校验格式为ogg json");

            //数据json方式处理
            JSONObject object = JSON.parseObject(data);
            //TODO 针对每个数据进行数据类型校验(待完善)
            String opType = (String) object.get("op_type");
            JSONObject beforObject = null;
            if (opType.trim().toLowerCase().equals("d")) {
                beforObject = object.getJSONObject("befor");

            } else {
                beforObject = object.getJSONObject("after");

            }
            if (beforObject != null) {
                for (int i = 0; i < dataKeys.size(); i++) {
                    Object dataObject = beforObject.get(dataKeys.get(i));
                    FlinkDataType.transformValue(FlinkDataType.getTypeInformation(dataTypes.get(i)).getTypeClass(),
                            (String) dataObject);

                }
            }

        } catch (Exception e) {
            logger.info("json数据==" + data);
            logger.info("json数据解析错误");
            return false;
        }
        return true;

    }
}
